package spring.cloud.tasks.client.item_task_processor;

import com.alibaba.fastjson.JSONObject;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import spring.cloud.tasks.client.ItemTaskWorker;
import spring.cloud.tasks.client.item_task_processor.pooled.TaskHeartBeatPooled;
import spring.cloud.tasks.client.item_task_processor.pooled.TaskPauseOrResumeScheduleTaskPooled;
import spring.cloud.tasks.client.item_task_processor.pooled.impl.TaskPauseOrResumeScheduleTaskPooledImpl;
import spring.cloud.tasks.client.item_task_processor.pooled.timer_task.PauseOrResumeScheduleTask;
import spring.cloud.tasks.client.item_task_processor.task_item.ClientTaskItemContractExtensionInfo;
import spring.cloud.tasks.client.util.CronExpression;
import spring.cloud.tasks.common.utils.ApplicationContextUtils;
import spring.cloud.tasks.common.utils.Uuids;
import spring.cloud.tasks.common.zookeeper.ZooKeeperDataManager;
import spring.cloud.tasks.common.zookeeper.ZooKeeperItemTaskDataManager;
import spring.cloud.tasks.common.zookeeper.ZookeeperNodeConstants;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

@Slf4j

public class TaskProcessor {
    @Setter
    @Getter
    private final Object needReloadTaskItemIdAndParameterListLock = new Object();
    @Setter
    @Getter
    public volatile ZooKeeperItemTaskDataManager zooKeeperItemTaskDataManager;
    @Setter
    @Getter
    public volatile boolean needPongPong = true;
    //在休眠模式下,任务执行过程中评估来当前任务执行不及休眠下,会和下次任务调度时间可能很近则不需要销毁自己的资源,一直执行下去,直到任务评估可以进行销毁的时候。
    public volatile boolean needReInitProcessor = true;
    @Setter
    @Getter
    public Date nextPauseTime;
    @Setter
    @Getter
    protected volatile Task task;
    @Setter
    @Getter
    protected volatile TaskProcessorInfo taskProcessorInfo;
    @Setter
    @Getter
    protected volatile String threadId;
    @Setter
    @Getter
    protected volatile String threadRuntimeErrorInfo = null;//调度器主线程运行时的异常
    @Setter
    @Getter
    protected volatile Lock processorLock = new ReentrantLock();
    @Setter
    @Getter
    protected volatile int taskItemListSizeInZooKeeper = 0;//所有的taskItemCount
    //@Setter
    //@Getter 存在同名的方法
    protected volatile List<String> taskItemList = new CopyOnWriteArrayList<String>();
    //
    @Setter
    @Getter
    protected volatile boolean stop = false;
    @Setter
    @Getter
    protected volatile boolean exceptionOccurred = false;//发生异常<自标记>
    @Setter
    @Getter
    protected volatile ConcurrentHashMap<String, ClientTaskItemContractExtensionInfo> taskItemToClientTaskItemContractExtensionInfoMap = new ConcurrentHashMap<String, ClientTaskItemContractExtensionInfo>();
    @Setter
    @Getter
    protected volatile ReentrantLock lock = new ReentrantLock();
    @Setter
    @Getter
    volatile ItemTaskWorker itemTaskWorker;
    //
    @Setter
    @Getter
    volatile TaskProcessingThreadContext taskProcessingThreadContext;
    @Setter
    @Getter
    volatile spring.cloud.tasks.client.item_task_processor.api.TaskProcessor taskProcessor;
    volatile TaskProcessingStatistics taskProcessingStatistics = new TaskProcessingStatistics();
    @Setter
    @Getter
    volatile boolean pause = true;
    @Setter
    @Getter
    volatile String pauseMessage = "";
    @Setter
    @Getter
    volatile String runMessage = "";
    @Setter
    @Getter
    volatile boolean scheduleTask = false;
    @Setter
    @Getter
    private volatile int threadCountPerTaskProcessor = 5;
    @Setter
    @Getter
    private volatile Date nextRunTime;
    //
    private TaskHeartBeatPooled taskHeartBeatPooled;
    private TaskPauseOrResumeScheduleTaskPooled taskPauseOrResumeScheduleTaskPooled;
    private PauseOrResumeScheduleTask resumeScheduleTask;
    private PauseOrResumeScheduleTask pauseScheduleTask;

    public TaskProcessor(String taskId, ItemTaskWorker itemTaskWorker, ZooKeeperDataManager zooKeeperItemTaskDataManager, int threadCountPerTaskProcessor) throws Exception {
        log.info("create TaskProcessor for task:" + taskId);
        this.task = this.zooKeeperItemTaskDataManager.loadTask(taskId);
        Object bean = ApplicationContextUtils.getApplicationContext().getBean(TaskIdToBeanIdFactory.getBeanId(taskId));
        this.taskProcessor = (spring.cloud.tasks.client.item_task_processor.api.TaskProcessor) bean;
        //
        this.itemTaskWorker = itemTaskWorker;
        this.zooKeeperItemTaskDataManager = (ZooKeeperItemTaskDataManager) zooKeeperItemTaskDataManager;
        this.threadCountPerTaskProcessor = threadCountPerTaskProcessor;
        this.taskProcessorInfo = createItemTaskProcessorInfo(taskId, itemTaskWorker, this.zooKeeperItemTaskDataManager, threadCountPerTaskProcessor);
        this.zooKeeperItemTaskDataManager.registerProcessor(this.taskProcessorInfo);
        newThread();
    }

    public static TaskProcessorInfo createItemTaskProcessorInfo(String taskId, ItemTaskWorker itemTaskWorker, ZooKeeperItemTaskDataManager zooKeeperItemTaskDataManager, int threadNum) {
        TaskProcessorInfo taskProcessorInfo = new TaskProcessorInfo();
        try {
            taskProcessorInfo.ip = InetAddress.getLocalHost().getHostAddress();
        } catch (UnknownHostException e) {
            e.printStackTrace();
        }
        taskProcessorInfo.taskId = taskId;
        taskProcessorInfo.workerId = itemTaskWorker.getWorkerId();
        taskProcessorInfo.processorId = taskId + ZookeeperNodeConstants.$ + itemTaskWorker.getWorkerId() + ZookeeperNodeConstants.$ + (Uuids.getUuid());//主要是uuid可能有问题
        taskProcessorInfo.threadNum = threadNum;
        taskProcessorInfo.registerTime = new Date(zooKeeperItemTaskDataManager.zooKeeperCurrentTimeMillis());
        taskProcessorInfo.heartBeatTime = null;
        taskProcessorInfo.message = "";
        taskProcessorInfo.version = 0;
        return taskProcessorInfo;
    }

    public void newThread() throws Exception {
        threadId = TaskProcessor.class.getName() + "-" + this.taskProcessorInfo.processorId;
        Thread thread = new Thread(threadId) {

            public void run() {
                try {
                    log.info(taskProcessorInfo.getProcessorId() + "开始启动");
                    while (true) {
                        if (stop == true) {
                            log.warn("停止调度器:" + taskProcessorInfo.getProcessorId());
                            return;
                        }
                        refreshProcessorInfo();
                        //再次被进行重新初始化
                        boolean taskItemsIsInited = zooKeeperItemTaskDataManager.taskItemsIsInited(task.getTaskId());
                        if (!taskItemsIsInited) {
                            TimeUnit.SECONDS.sleep(10);//存在分配task item ，需要等待stop被设置为true.
                            continue;
                        }
                        if (getTaskItemList().size() > 0) {
                            taskItemListSizeInZooKeeper = zooKeeperItemTaskDataManager.loadTaskItemList(taskProcessorInfo.getTaskId()).size();
                            break;
                        }
                        TimeUnit.SECONDS.sleep(1);
                    }
                    log.info("调度器:" + taskProcessorInfo.getProcessorId() + "拉取任务:" + JSONObject.toJSONString(taskItemList));
                    schedule();
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                    threadRuntimeErrorInfo = e.getMessage();
                    exceptionOccurred = true;//出现异常
                }
            }//异步
        };
        thread.setDaemon(true);
        thread.start();
    }

    public void refreshProcessorInfo() throws Exception {
        processorLock.lock();
        try {
            if (this.stop == true) {
                return;
            }
            StringBuilder sb = new StringBuilder();
            sb.append("暂停信息:");
            if (pauseMessage != null) {
                sb.append(pauseMessage);
            } else {
                sb.append("-");
            }
            sb.append(";");
            sb.append("线程运行时异常:");
            if (threadRuntimeErrorInfo != null) {
                sb.append(threadRuntimeErrorInfo);
            } else {
                sb.append("-");
            }
            sb.append(";");
            this.taskProcessorInfo.setMessage(sb.toString());
            //
            if (!this.zooKeeperItemTaskDataManager.updateProcessorHeartBeatTimeAndVersion(this.taskProcessorInfo)) {//可能异常
                this.clearTaskItemAndClearTaskDataIdList();
            }
        } catch (Exception e) {
            exceptionOccurred = true;
        } finally {
            processorLock.unlock();
        }
    }

    public void unRegisterProcessor() throws Exception {
        processorLock.lock();
        try {
            this.zooKeeperItemTaskDataManager.unRegisterProcessor(this.taskProcessorInfo);
        } finally {
            processorLock.unlock();
        }
    }

    /**
     * 守护线程调用
     * 启动一个processor  如果processor 正在处理中 则跳过这次启动
     */
    public boolean startProcessor() throws Exception {
        boolean reGetTaskItemList = true;
        String message = "";
        return resume(reGetTaskItemList, message);
    }

    /**
     * 守护线程调用-停止处理
     */
    public boolean stopProcessor() throws Exception {
        String message = "";
        return pause(message);
    }

    /**
     * 重启,该功能可以进行任务处理器初始化
     */
    public synchronized boolean resume(boolean reGetTaskItemList, String message) throws Exception {
        if (this.isStop()) {
            return false;
        }
        if (this.pause == true) {
            //
            refreshProcessorInfo();
            //
            this.pause = false;
            //
            this.runMessage = message;
            //
            this.taskProcessingThreadContext = new TaskProcessingThreadContext(this, taskProcessor, this.taskProcessingStatistics);
            if (reGetTaskItemList) {
                getTaskItemList();
            }
            return true;

        }
        return true;
    }

    /**
     * 暂停任务执行
     */
    public synchronized boolean pause(String message) throws Exception {
        if (!this.pause) {
            this.pause = true;
            this.pauseMessage = message;
            //
            if (this.taskProcessingThreadContext != null) {
                this.taskProcessorInfo.setState(ProcessorStatusEnum.pause.toString());
                this.taskProcessingThreadContext.destroy();
                this.taskProcessingThreadContext = null;
            }
        }
        return true;
    }

    public void stop() throws Exception {
        needPongPong = false;//不再进行心跳
        //
        this.pause = true;//用于控制调度器是否暂停
        this.stop = true;//用于控制调度器是否停止
        //
        if (this.taskProcessingThreadContext != null) {
            this.taskProcessingThreadContext.destroy();
            this.taskProcessingThreadContext = null;
        }
        //
        this.unRegisterProcessor();

        this.unRegisterPauseOrResumeScheduleTask(this.taskPauseOrResumeScheduleTaskPooled);
        this.unRegisterHeartBeatTask(this.taskHeartBeatPooled);

    }

    /**
     * 任务执行完毕后才进行拉取数据
     */
    public List<String> getTaskItemListUntilTaskDataListIsAllFinshed() {
        try {

            if (this.taskProcessingThreadContext != null) {
                while (!this.taskProcessingThreadContext.noTaskDatas()) {
                    TimeUnit.SECONDS.sleep(1);
                }//must all task is all finshed
            }
            synchronized (needReloadTaskItemIdAndParameterListLock) {
                this.getTaskItemList();

            }
            return this.taskItemList;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    protected synchronized List<String> getTaskItemList() throws Exception {

        try {
            this.taskItemList.clear();//不能处理可能无效的task item
            this.zooKeeperItemTaskDataManager.releaseContractExtensionInfos(this.taskProcessorInfo, taskItemToClientTaskItemContractExtensionInfoMap);
            //
            zooKeeperItemTaskDataManager.getContractExtensionInfos(taskProcessorInfo, taskItemToClientTaskItemContractExtensionInfoMap);
            this.taskItemList = this.zooKeeperItemTaskDataManager.getTaskItemAndParameterList(this.taskProcessorInfo);
            return this.taskItemList;
        } catch (Throwable e) {

            if (e instanceof Exception) {
                throw (Exception) e;
            } else {
                throw new Exception(e);
            }
        }
    }

    public void clearTaskItemAndClearTaskDataIdList() {
        this.taskItemList.clear();
        if (this.taskProcessingThreadContext != null) {
            this.taskProcessingThreadContext.clearTaskDataIdList();
        }
    }

    public void schedule() throws Exception {
        boolean startRunMode = false;
        if (this.task.getRunCronExpressionString() == null) {
            startRunMode = true;
        } else {
            String cronExpressionString = this.task.getRunCronExpressionString();
            if (cronExpressionString.toLowerCase().startsWith("rightNowRun:")) {
                cronExpressionString = cronExpressionString.substring("rightNowRun:".length());
                this.task.setRunCronExpressionString(cronExpressionString);
                startRunMode = true;
            }
            CronExpression cronExpression = new CronExpression(cronExpressionString);
            Date nowDate = new Date(this.zooKeeperItemTaskDataManager.zooKeeperCurrentTimeMillis());
            Date nextRunTime = cronExpression.getNextTimeAfter(nowDate);
            this.nextRunTime = nextRunTime;
            this.taskProcessorInfo.setNextRunTime(nextRunTime);
            //
            //this.heartBeatTimer.schedule(new TaskProcessorScheduleTimerTask(this, this.heartBeatTimer, cronExpressionString), nextRunTime);
            this.scheduleTask = true;//外部调度定期器可以真正进行调度
        }
        if (startRunMode == true) {
            this.resume(false, "");
        }
    }

    public boolean isContinueWhenNoTaskDataIdList() throws Exception {
        if (this.task.getRunCronExpressionString() == null) {
            this.needReInitProcessor = false;
            return true;//如果没有外部命令永远没有清理的机会
        } else {
            if (this.taskItemList.size() > 0) {
                this.needReInitProcessor = false;
                return true;
            } else {
                //进行比较现在的时间和下次运行的时间是否很近
                boolean close = (nextRunTime.getTime() - (new Date()).getTime()) <= 1000 * 60;
                if (close) {
                    this.needReInitProcessor = false;
                    return true;
                } else {
                    //this.taskItemList.size()==0
                    this.needReInitProcessor = true;
                    this.pause("");
                    return false;
                }

            }
        }

    }

    public void pongpong() throws Exception {
        try {
            refreshProcessorInfo();
        } catch (Throwable e) {
            exceptionOccurred = true;
            this.clearTaskItemAndClearTaskDataIdList();
            //如果出现了异常需要进行异常以及新的调度器分配
            if (e instanceof Exception) {
                throw (Exception) e;
            } else {
                throw new Exception(e.getMessage(), e);
            }
        }
    }


    public void registerHeartBeatTask(TaskHeartBeatPooled taskHeartBeatService) {
        this.taskHeartBeatPooled = taskHeartBeatService;
        this.taskHeartBeatPooled.add(this.task.getHeartBeatRate(), this);
    }


    public void unRegisterHeartBeatTask(TaskHeartBeatPooled taskHeartBeatService) {
        this.taskHeartBeatPooled = taskHeartBeatService;
        this.taskHeartBeatPooled.remove(this.task.getHeartBeatRate(), this);
    }

    public void registerPauseOrResumeScheduleTask(TaskPauseOrResumeScheduleTaskPooled pauseOrResumeScheduleTaskService) {
        this.taskPauseOrResumeScheduleTaskPooled = pauseOrResumeScheduleTaskService;

        Date nowDate = new Date(this.zooKeeperItemTaskDataManager.zooKeeperCurrentTimeMillis());
        Date nextRunTime = (new CronExpression(this.task.getRunCronExpressionString())).getNextTimeAfter(nowDate);
        this.taskProcessorInfo.setNextRunTimeString(nextRunTime.toString());
        this.taskProcessorInfo.setNextRunTime(nextRunTime);
        this.nextRunTime = nextRunTime;
        //
        this.resumeScheduleTask = new PauseOrResumeScheduleTask(this, taskProcessorInfo, PauseOrResumeScheduleTask.TYPE_RESUME, this.task.getRunCronExpressionString());
        this.taskPauseOrResumeScheduleTaskPooled.init(TaskPauseOrResumeScheduleTaskPooledImpl.resume, this.resumeScheduleTask, nextRunTime);
        //
        if (this.task.getPauseCronExpressionString() != null) {
            Date nextPauseTime = new CronExpression(this.task.getPauseCronExpressionString()).getNextTimeAfter(nextRunTime);
            this.taskProcessorInfo.setNextPauseTimeString(nextPauseTime.toString());
            this.taskProcessorInfo.setNextPauseTime(nextPauseTime);
            this.nextPauseTime = nextPauseTime;
            //
            this.pauseScheduleTask = new PauseOrResumeScheduleTask(this, taskProcessorInfo, PauseOrResumeScheduleTask.TYPE_PAUSE, this.task.getPauseCronExpressionString());
            this.taskPauseOrResumeScheduleTaskPooled.init(TaskPauseOrResumeScheduleTaskPooledImpl.pause, this.pauseScheduleTask, nextPauseTime);

        }
    }


    public void unRegisterPauseOrResumeScheduleTask(TaskPauseOrResumeScheduleTaskPooled pauseOrResumeScheduleTaskService) {
        this.taskPauseOrResumeScheduleTaskPooled = pauseOrResumeScheduleTaskService;
        if (this.resumeScheduleTask != null) {
            this.taskPauseOrResumeScheduleTaskPooled.cancel(TaskPauseOrResumeScheduleTaskPooledImpl.resume, this.resumeScheduleTask);
        }
        if (this.pauseScheduleTask != null) {
            this.taskPauseOrResumeScheduleTaskPooled.cancel(TaskPauseOrResumeScheduleTaskPooledImpl.pause, this.pauseScheduleTask);
        }
    }

}

